package com.xxx.rpc.client;

import java.io.IOException;
import java.util.UUID;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Future;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.xxx.rpc.common.bean.RpcRequest;
import com.xxx.rpc.common.bean.RpcResponse;
import com.xxx.rpc.common.codec.RpcDecoder;
import com.xxx.rpc.common.codec.RpcEncoder;
import com.xxx.rpc.common.util.Global;
import com.xxx.rpc.context.RpcContext;
import com.xxx.rpc.registry.ServiceDiscovery;

import io.netty.bootstrap.Bootstrap;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelHandler.Sharable;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandler;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelOption;
import io.netty.channel.ChannelPipeline;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.SimpleChannelInboundHandler;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;

/**
 * RPC 客户端（用于发送 RPC 请求）
 *
 * @author huangyong & Rosun
 * @since 1.0.0
 */
@Sharable
public class NettyRpcClient extends SimpleChannelInboundHandler<RpcResponse> implements RpcClient {

	private static final Logger LOGGER = LoggerFactory.getLogger(NettyRpcClient.class);

	// private ChannelHandlerContext ctx;
	private Channel channel;

	private EventLoopGroup group = new NioEventLoopGroup(1);
	
	private ServiceDiscovery point;

	private static ExecutorService threadPool = Global.getThreadPool();
	
	private String id = null;

	public NettyRpcClient() {
		this.id = UUID.randomUUID().toString();
	}

	@Override
	public void channelActive(ChannelHandlerContext ctx) throws Exception {
		this.channel = ctx.channel();
	}

	/**
	 * Calls {@link ChannelHandlerContext#fireChannelInactive()} to forward to
	 * the next {@link ChannelInboundHandler} in the {@link ChannelPipeline}.
	 *
	 * Sub-classes may override this method to change behavior.
	 */
	@Override
	public void channelInactive(ChannelHandlerContext ctx) throws Exception {
		// super.channelInactive(ctx);
		LOGGER.error("channelInactive channelInactive channelInactive");
		// 0 通知服务发现接口，本链接已经断球了，不能再用了；
		if(this.point != null){
			this.point.rpcClientCloseCallBack(this);
			this.point = null;
		}
		// 1 告警
		this.warningChannelInActive();
		// 2 资源回收
		this.destroy();
		// 3 重新连接；
		this.group = new NioEventLoopGroup();
		while (this.channel == null) {
			LOGGER.info("retry connect to server , " + host + ":" + port);
			try{
				this.initLongConnection(host, port);
			}catch(Exception ex){
				LOGGER.error("retry initLongConnection error: {},{}",ex.getMessage(),ex.getClass().getName());
				Thread.sleep(5 * 1000);
			}
		}
		//处理完毕！
		LOGGER.info("process channelInactive,create new connection channel:" + this.channel.isActive());
	}

	private void warningChannelInActive() {
		System.err.println("channelInactive  channelInactive  channelInactive ");
		System.err.println("channelInactive  channelInactive  channelInactive ");
		System.err.println("channelInactive  channelInactive  channelInactive ");
	}

	@Override
	public void channelRead0(ChannelHandlerContext ctx, RpcResponse response) throws Exception {
		// this.response = response;
		final String id = response.getRequestId();
		RpcRequest request = RpcContext.getReqMap().remove(id);
		request.getCallBack().doCallBack(response);
	}

	@Override
	public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
		LOGGER.error("api caught exception,{},{}", cause.getMessage(), cause.getClass().getName());
		// ctx.close(); //FIXME 注释掉 by罗珊
		// //告警? 采集 ？
	}

	// 缓存起来以便重新连接；
	private String host;
	private int port;

	public void initLongConnection(String host, int port) throws Exception {
		// 缓存起来以便重新连接；
		this.host = host;
		this.port = port;
		// 缓存起来以便重新连接；
		try {
			// 创建并初始化 Netty 客户端 Bootstrap 对象
			Bootstrap bootstrap = new Bootstrap();
			
			bootstrap.channel(NioSocketChannel.class);
			bootstrap.option(ChannelOption.TCP_NODELAY, true);
			bootstrap.option(ChannelOption.SO_KEEPALIVE, true);
			bootstrap.remoteAddress(host, port);
			bootstrap.group(group);		
			bootstrap.handler(new ChannelInitializer<SocketChannel>() {
				@Override
				public void initChannel(SocketChannel channel) throws Exception {
					ChannelPipeline pipeline = channel.pipeline();
					// 客户端：
					pipeline.addLast(new RpcEncoder(RpcRequest.class)); // 编码 RPC  请求
					pipeline.addLast(new RpcDecoder(RpcResponse.class)); // 解码 RPC 响应

					// 服务端：
					// pipeline.addLast(new RpcDecoder(RpcRequest.class)); // 解码 RPC 请求
					// pipeline.addLast(new RpcEncoder(RpcResponse.class)); // 编码 RPC 响应
					pipeline.addLast(NettyRpcClient.this); // 处理 RPC 响应
				}
			});

			// 连接 RPC 服务器
			ChannelFuture future = bootstrap.connect(host, port).sync();
			if (future.isSuccess()) {
				// 写入 RPC 请求数据并关闭连接
				this.channel = future.channel();
				LOGGER.info("#### initLongConnection successfuly connectet to " + host + "@" + port);
			} else {
				LOGGER.error("#### initLongConnection failed connectet to " + host + "@" + port);
				throw new IllegalStateException("无法连接到服务器：" + host + ":" + port);
			}
			// channel.writeAndFlush(request).sync();
			// channel.closeFuture().sync();
		} catch (Exception ex) {
			LOGGER.error(ex.getMessage());
			throw ex;
		} finally {

		}
	}

	public void closeLongConnection() {
		try {
			this.group.shutdownGracefully();
		} catch (Exception ex) {
			LOGGER.error(ex.getMessage(), ex);
		} finally{
			this.group = null;
		}
	}

	public RpcResponse send(final RpcRequest request) throws Exception {
		return send(request,0);
	}
	
	public RpcResponse send(final RpcRequest request,final long timeout) throws Exception {
		if(this.channel == null || !this.channel.isActive()){
			RpcResponse resp = new RpcResponse();
			resp.setException(new IOException("channel is NOT active, request ignored:"+ request));
			resp.setRequestId(request.getRequestId());
			return resp;
		}
		
		final CallBack cb = new DefauldCallBack();
		request.setCallBack(cb);
		RpcContext.getReqMap().put(request.getRequestId(), request);

		Callable<RpcResponse> job = new Callable<RpcResponse>() {
			@Override
			public RpcResponse call() throws Exception {
				
				try {
					// cb.doSendThroughChannel(ctx.channel(), request);
					cb.doSendThroughChannel(channel, request);
				} catch (Exception ex) {
					//FIXME 告警？？？ 消息发送不出去了！！！
					RpcResponse resp = new RpcResponse();
					resp.setException(ex);
					resp.setRequestId(request.getRequestId());
					return resp;
				}
				if(timeout <= 0){
					return cb.get(request.getRequestId());
				}else{
					return cb.get(request.getRequestId(),timeout);
				}
			}
		};

		Future<RpcResponse> future = threadPool.submit(job);
		// 返回 RPC 响应对象
		RpcResponse resp = future.get();
		if(resp.isTimeOut()){
			LOGGER.error("$$$$$$$  TIME OUT EXCEPTION :",resp.getException());
			LOGGER.error("$$$$$$$  TIME OUT EXCEPTION ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^\n\n");
		}
		//System.out.println("future.get() ok ！！" + resp);
		return resp;
	}
	

	@Override
	public void destroy() {
		try {
			// if(this.ctx != null){
			// this.ctx.close();
			// this.ctx = null;
			// }
			if (this.channel != null) {
				this.channel.close();
				this.channel = null;
			}
		} catch (Exception ex) {
			LOGGER.error(ex.getMessage(), ex);
		}

		this.closeLongConnection();
	}

	@Override
	public void setServiceDiscoveryPoint(ServiceDiscovery point) {
		this.point = point;
	}

	/**
	 * @return the id
	 */
	public String getId() {
		return id;
	}

	public String toString(){
		return "RpcClient:"+this.id;
	}
	
	/**
	 * (non-Javadoc)
	 * @see java.lang.Object#hashCode()
	 */
	@Override
	public int hashCode() {
		final int prime = 31;
		int result = 1;
		result = prime * result + ((id == null) ? 0 : id.hashCode());
		return result;
	}

	/**
	 * (non-Javadoc)
	 * @see java.lang.Object#equals(java.lang.Object)
	 */
	@Override
	public boolean equals(Object obj) {
		if (this == obj) {
			return true;
		}
		if (obj == null) {
			return false;
		}
		if (!(obj instanceof NettyRpcClient)) {
			return false;
		}
		NettyRpcClient other = (NettyRpcClient) obj;
		if (id == null) {
			if (other.id != null) {
				return false;
			}
		} else if (!id.equals(other.id)) {
			return false;
		}
		return true;
	}
}
